package com.lfd.rocketmq.service;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;

/**
 * @项目名： spring-emq
 * @包名： com.lfd.rocketmq.service
 * @文件名: ConsumerRockMQ
 * @创建者: liufudong@szyh-smart.com
 * @创建时间: 2024/9/30 21:50
 * @描述：
 */
@Slf4j
@Service
public class ConsumerProducerRockMQ {
    @Autowired
    private ObjectMapper mapper;
    /**
     * @param consumerGroup 如"ConsumerRockMQ"
     * @param namesrvAddr   如"127.0.0.1:9876"
     * @param var1 订阅主题
     * @param var2 标签
     * @throws MQClientException
     */
    public void registerMessageListener(String consumerGroup, String namesrvAddr,
                                        String var1, String var2) throws MQClientException {
        // 创建消费者并指定消费者组
        DefaultMQPushConsumer consumerRockMQ =
                new DefaultMQPushConsumer(consumerGroup);
        // 设置 NameServer 地址
        consumerRockMQ.setNamesrvAddr(namesrvAddr);
        // 订阅主题和标签
        consumerRockMQ.subscribe(var1, var2);
        // 注册消息监听器 MessageListenerConcurrently 是多线程消费，默认20个线程，可以参看consumer.setConsumeThreadMax()
        consumerRockMQ.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt msg : list) {
                    // 获取消息体的字节数组
                    byte[] body = msg.getBody();
                    // 将字节数组转换为字符串
                    String messageBody = new String(body);

                    try {
                        log.info("ConsumerRockMQ  msg: {}", mapper.writeValueAsString(msg));
                        log.info("ConsumerRockMQ  接受的消息: {}", mapper.writeValueAsString(messageBody));
                    } catch (JsonProcessingException e) {
                        throw new RuntimeException(e);
                    }
                }
                // 消息处理成功
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                //发生了失败或错误。
//                return  ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        });
        //启动消费者
        consumerRockMQ.start();
        log.info("ConsumerRockMQ started");
    }
}
